package com.example.mq.server.datacenter;

import com.example.mq.common.MqException;
import com.example.mq.server.core.Binding;
import com.example.mq.server.core.Exchange;
import com.example.mq.server.core.MSGQueue;
import com.example.mq.server.core.Message;
import lombok.extern.slf4j.Slf4j;

import java.io.IOException;
import java.util.LinkedList;
import java.util.List;
import java.util.concurrent.ConcurrentHashMap;

/*
* 使用这个对象来管理内存中的数据
* 该类后续的一些方法可能会在多线程的环境下被调用,因此需要注意线程安全问题
* */
@Slf4j
public class MemoryDataCenter {
    // key = exchangeName   value = exchange对象
    private final ConcurrentHashMap<String, Exchange> exchangeMap = new ConcurrentHashMap<>();
    // key = queueName      value = MsgQueue对象
    private final ConcurrentHashMap<String, MSGQueue> queueMap = new ConcurrentHashMap<>();
    // 第一个key 是exchangeName   第二个key是queueName
    private final ConcurrentHashMap<String,ConcurrentHashMap<String, Binding>> bindingsMap = new ConcurrentHashMap<>();
    // key = messageId      value = message对象
    private final ConcurrentHashMap<String, Message> messageMap = new ConcurrentHashMap<>();
    // key = queueName      value = message链表
    private final ConcurrentHashMap<String, LinkedList<Message>> queueMessageMap = new ConcurrentHashMap<>();
    // 第一个key是队列名称  第二个key是消息id
    private final ConcurrentHashMap<String,ConcurrentHashMap<String,Message>> queueMessageWaitAckMap = new ConcurrentHashMap<>();


    /**
     * 插入交换机
     * @param exchange
     */
    public void insertExchange(Exchange exchange){
        exchangeMap.put(exchange.getName(),exchange);
        log.info("[MemoryDataCenter] 新交换机添加成功! exchangeName = "+exchange.getName());
    }

    /**
     * 取交换机
     * @param exchangeName
     * @return
     */
    public Exchange getExchange(String exchangeName){
        return exchangeMap.get(exchangeName);
    }

    /**
     * 删除交换机
     * @param exchangeName
     */
    public void deleteExchange(String exchangeName){
        exchangeMap.remove(exchangeName);
        log.info("[MemoryDataCenter] 交换机删除成功! exchangeName = "+exchangeName);
    }

    /**
     * 插入队列
     * @param queue
     */
    public void insertQueue(MSGQueue queue){
        queueMap.put(queue.getName(),queue);
        log.info("[MemoryDataCenter] 新队列添加成功! queueName = "+queue.getName());
    }

    /**
     * 取队列
     * @param queueName
     * @return
     */
    public MSGQueue getQueue(String queueName){
        return queueMap.get(queueName);
    }

    /**
     * 删除交换机
     * @param queueName
     */
    public void deleteQueue(String queueName){
        queueMap.remove(queueName);
        log.info("[MemoryDataCenter] 队列删除成功! queueName = "+queueName);
    }

    /**
     * 建立绑定
     * @param binding
     */
    public void insertBinding(Binding binding) throws MqException {
        // 1.先试用exchangeName查一下哈希表是否存在
/*        ConcurrentHashMap<String, Binding> bindingMap = bindingsMap.get(binding.getExchangeName());
        if(bindingMap == null){
            bindingMap = new ConcurrentHashMap<>();
            bindingsMap.put(binding.getExchangeName(),bindingMap);
        }*/
        ConcurrentHashMap<String,Binding> bindingMap = bindingsMap.computeIfAbsent(binding.getExchangeName(),
                k -> new ConcurrentHashMap<>());

        // 再根据queueName查一下,如果已经存在,抛出异常
        // get put 联动 加锁
        synchronized (bindingMap) {
            Binding binding1 = bindingMap.get(binding.getQueueName());
            if(binding1 != null){
                throw new MqException("[MemoryDataCenter] 绑定已经存在 queueName = "+binding.getQueueName()
                        +" ExchangeName = "+binding.getExchangeName());
            }
            bindingMap.put(binding.getQueueName(),binding);
        }
        log.info("[MemoryDataCenter] 绑定建立成功! exchangeName = "+binding.getExchangeName()+" queueName = "+binding.getQueueName());
    }

    // 单个binding
    public Binding getBinding(String exchangeName,String queueName){
        ConcurrentHashMap<String, Binding> bindingMap = bindingsMap.get(exchangeName);
        if(bindingMap == null){
            return null;
        }
        return bindingMap.get(queueName);
    }

    // 多个binding
    public ConcurrentHashMap<String,Binding> getBinding(String exchangeName){
        return bindingsMap.get(exchangeName);
    }

    // 删除绑定
    public void deleteBinding(Binding binding) throws MqException {
        ConcurrentHashMap<String, Binding> bindingMap = bindingsMap.get(binding.getExchangeName());
        if(bindingMap == null){
            // 交换机没有绑定任何队列
            throw new MqException("[MemoryDataCenter] 交换机没有绑定任何队列 exchangeName = "+binding.getExchangeName()+" queueName = "+binding.getQueueName());
        }
        bindingMap.remove(binding.getQueueName());
        log.info("[MemoryDataCenter] 绑定删除成功! exchangeName = "+binding.getExchangeName()+" queueName = "+binding.getQueueName());
    }

    public void addMessage(Message message){
        messageMap.put(message.getMessageId(),message);
        log.info("[MemoryDataCenter] 新消息添加成功! messageId = "+message.getMessageId());
    }

    public Message getMessage(String messageId){
        return messageMap.get(messageId);
    }

    public void removeMessage(String messageId){
        messageMap.remove(messageId);
        log.info("[MemoryDataCenter] 消息移除成功! messageId = "+messageId);
    }

    // 发送消息到指定队列
    public void sendMessage(MSGQueue queue,Message message){
        LinkedList<Message> messageLinkedList = queueMessageMap.computeIfAbsent(queue.getName(), k -> new LinkedList<>());
        messageLinkedList.add(message);
    }

    // 从队列中取消息
    public Message pollMessage(MSGQueue queue) {
        LinkedList<Message> messageLinkedList = queueMessageMap.get(queue.getName());
        if(messageLinkedList == null){
            return null;
        }

        synchronized (messageLinkedList) {
            if(messageLinkedList.size() == 0){
                return null;
            }
            // 链表中有元素进行头删
            Message currentMessage = messageLinkedList.remove(0);
            log.info("[MemoryDataCenter] 消息从队列中取出 messageId = "+currentMessage.getMessageId());
            return currentMessage;
        }
    }

    // 获取队列中消息的个数
    public int getMessageCount(String queueName){
        LinkedList<Message> messageLinkedList = queueMessageMap.get(queueName);
        if(messageLinkedList == null){
            return 0;
        }
        synchronized (messageLinkedList){
            return messageLinkedList.size();
        }
    }

    // 添加未确认的消息
    public void addMessageWaitAck(String queueName,Message message){
/*        ConcurrentHashMap<String, Message> messages = queueMessageWaitAckMap.get(queueName);
        if(messages == null){
            messages = new ConcurrentHashMap<>();
            queueMessageWaitAckMap.put(queueName,messages);
        }*/
        ConcurrentHashMap<String, Message> messages = queueMessageWaitAckMap.computeIfAbsent(queueName,
                k -> new ConcurrentHashMap<>());
        messages.put(message.getMessageId(),message);
        log.info("[MemoryDataCenter] 添加未确认消息成功! messageId = "+message.getMessageId());
    }

    // 删除未确认的消息
    public void removeMessageWaitAck(String queueName,String messageId){
        ConcurrentHashMap<String, Message> messages = queueMessageWaitAckMap.get(queueName);
        if(messages == null){
            return;
        }
        messages.remove(messageId);
        log.info("[MemoryDataCenter] 删除未确认消息成功! messageId = "+messageId);
    }

    // 获取指定未确认的消息
    public Message getMessageWaitAck(String queueName,String messageId) throws MqException {
        ConcurrentHashMap<String, Message> messages = queueMessageWaitAckMap.get(queueName);
        if(messages == null){
            return null;
        }
        return messages.get(messageId);
    }

    // 将硬盘中持久化的数据恢复到内存中
    public void recovery(DiskDataCenter diskDataCenter) throws MqException, IOException, ClassNotFoundException {
        // 先清空数据
        exchangeMap.clear();
        queueMap.clear();
        bindingsMap.clear();
        messageMap.clear();
        queueMessageMap.clear();
        // 1.恢复所有的交换机数据
        List<Exchange> exchanges = diskDataCenter.selectAllExchanges();
        for(Exchange exchange : exchanges){
            exchangeMap.put(exchange.getName(),exchange);
        }

        // 2.恢复所有的队列数据
        List<MSGQueue> msgQueues = diskDataCenter.selectAllQueues();
        for (MSGQueue msgQueue : msgQueues) {
            queueMap.put(msgQueue.getName(),msgQueue);
        }

        // 3.恢复所有的绑定数据
        List<Binding> bindings = diskDataCenter.selectAllBindings();
        for (Binding binding : bindings) {
            ConcurrentHashMap<String, Binding> bindingList = bindingsMap.computeIfAbsent(binding.getExchangeName(),
                    k -> new ConcurrentHashMap<>());
            bindingList.put(binding.getQueueName(),binding);
        }

        // 4.恢复所有的消息数据
        for (MSGQueue queue : msgQueues) {
            LinkedList<Message> messages = diskDataCenter.loadAllMessageFromQueue(queue.getName());
            queueMessageMap.put(queue.getName(),messages);
            for (Message message : messages) {
                messageMap.put(message.getMessageId(),message);
            }
        }

    }
}

